package spark.core.scala

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by ibf on 2018/1/28.
 */
object PVandUVSpark {

  def main(args: Array[String]) {
    //1、创建spark上下文
    val conf = new SparkConf()
        .setMaster("local[*]")
        .setAppName("PVandUVSpark")
    val sc = new SparkContext(conf)

    //2、读取数据文件形成RDD
    val path = "datas/page_views.data"
    val rdd: RDD[String] = sc.textFile(path)

    //3、处理数据
    /**
     * 运行实际计算数据的spark代码的时候，往往之前会有
     * 几十个甚至上百个的ETL程序
     * 帮助我们将最原始的数据清洗成能够处理的spark数据
     *
     */
    val index = 10
    val len = 7
    val mapredRdd = rdd.map(line => line.split("\t"))
        .filter(arr => {
          //日志的长度必须等于7，并且date这个字段不能缺失
          arr.length == len && arr(0).trim.length > index
        })
        .map(arr => {
      //当前日志信息的日期（过滤掉小时）
      val date = arr(0).substring(0,index)
      val url = arr(1).trim
      val guid = arr(2).trim
      //最后返回一个三元组，将日期，url，guid做返回
     (date,url,guid)

    })

    //然后因为mapredRdd我们接下来要进行多次的操作
    //所以要cache
    mapredRdd.cache()

    /**
     * 计算PV
     * 只需要date和url
     * 所以一开始就可以用map将guid过滤掉
     */
    /*val pvRDD= mapredRdd.map(t => (t._1,t._2))
          //当每次做filter的时候记得考虑一个问题
      /**
       * filter过滤的数据会不会特别多
       * 如果过滤特别多的时候
       * 会出现某些分区，数据非常的小
       * 但是他却也需要占用1个task资源
       * 所以，当出现这种情况的时候，可以考虑重分区操作
       * 将数据做一个重新分区，让分区中的数据尽量均衡
       * repartition()
       * coalesce()
       * 留个小问题，大家自己去看下这两个方法区别？
       */
          .filter(t => t._2.nonEmpty)
          .groupByKey()
          .map(t => {
            val date: String = t._1
            val iter: Iterable[String] = t._2
            (date,iter.size)
          })*/
    /**
     * 优化：
     *  url你是否需要保留他的值？
     *  如果这个url从头到尾，我们都不需要知道他的值是多少，
     *  那么可以考虑，将url也过滤掉！
     */
    val pvRDD= mapredRdd
        .filter(t => t._2.nonEmpty)
      //直接将url转换成一个Int类型的1
        .map(t => (t._1,1))
        .reduceByKey(_ + _)


    /**
     * 计算UV
     *  guid或者session或者userid是有可能重复的，计算UV之类的值的时候
     *	考虑如何去重？
     *  在hive当中， distinct 	关键字
     *  方法一：使用Set（无序不重复）的集合帮我们做去重
     *        但是：当数据量太大的时候，这种方法明显不合理
     */

   /*val uvRDD: RDD[(String, Int)] =  mapredRdd.map(t => (t._1,t._3))
        .filter(_._2.nonEmpty)
      //将相同的key（date）的数据聚合到一起
      //iter  ==》 iter当中的数据是有重复
        .groupByKey()
        .map(t => {
        val date = t._1
        val iter: Iterable[String] = t._2
        (date,iter.toSet.size)
      })*/

    /**
     * 方法二：
     *  key可以是什么值？
     *  应该是相同天的相同的guid，我们认为他是同一条数据
     *  是否可以将数据做成这种格式   ((date,guid),1)
     */
   /* val uvRDD= mapredRdd.filter(t =>t._3.nonEmpty)
            .map(t => {
          val date = t._1
          val guid = t._3
          val value = 1
        ((date,guid),1)
      })
        //我们不需要得到topN，不需要做排序，所以并不用将每个数据的value做叠加
        .reduceByKey((a,b) => a)
        .map(t => {
        val key = t._1
        val count = t._2
        val date = key._1
        val guid = key._2
      //代表当天有一个guid访问！
      (date,1)
    })
    .reduceByKey(_ + _)
*/
    /**
     * 方法三： 其实RDD中有过滤的方法
     *    我们可以直接使用
     *
     */
    val uvRDD= mapredRdd.filter(t =>t._3.nonEmpty)
          .map(t => (t._1,t._3))
      //去重的方法
          .distinct()
//      .map(t => (t._1,1))
//      .reduceByKey(_ + _)

    //想要将PV和UV的数据做join（聚合）
    val finalResultRDD = pvRDD.fullOuterJoin(uvRDD)
      .map(t => {
      val date = t._1
      val pvOption = t._2._1
      val uvOption = t._2._2
      //如果能够聚合到pv或者uv的值，就返回正确的值，如果没有就返回-1
      (date,pvOption.getOrElse(-1),uvOption.getOrElse(-1))
    })


    //打印方法
    println("===========PV统计================")
    pvRDD.foreachPartition(itr => {
      itr.foreach(println)
    })
    println("===========UV统计================")
    uvRDD.foreachPartition(itr => {
      itr.foreach(println)
    })
    println("===========聚合统计================")
    finalResultRDD.foreachPartition(itr => {
      itr.foreach(println)
    })


    //释放cache的缓存
    mapredRdd.unpersist()

    //4、线程等待
    Thread.sleep(1000000l)

  }
}
